package de.mmis.core.tuplespace;

import java.util.concurrent.TimeoutException;

import de.mmis.core.base.abstracttree.InnerNode;
import de.mmis.core.base.abstracttree.LeafNode;
import de.mmis.core.base.abstracttree.Tree;

public class TuplespaceEventQueue extends TuplespaceQueue {
	private long nextEventIndexToRead = Long.MAX_VALUE;

	public TuplespaceEventQueue(Tuplespace space, String name, Tree id)
			throws TuplespaceException {
		super(space, name, id);
	}

	public TuplespaceEventQueue(Tuplespace space, String name, Tree id,
			long timeout) throws TuplespaceException {
		super(space, name, id, timeout);
	}

	public TuplespaceEventQueue(Tuplespace space, String name, Tree id,
			long timeout, long leaseTime) throws TuplespaceException {
		super(space, name, id, timeout, leaseTime);
	}

	public void initEventQueueReader() throws TuplespaceException,
			TimeoutException {
		initEventQueueReader(timeout);
	}

	public void initEventQueueReader(long timeout) throws TuplespaceException,
			TimeoutException {
		InnerNode queue = (InnerNode) space.read(queueQueryTuple, timeout);
		if (queue == null)
			throw new TimeoutException("queue tuple read timed out");

		// set next event index to read
		long end = Long.parseLong(((LeafNode) queue.getSubTree(3))
				.getUTF8Data());
		nextEventIndexToRead = end + 1;

		// System.out.println("EventQueueReader initialized, nextEventIndexToRead = "+nextEventIndexToRead);
	}

	public Tree readNextEvent() throws TuplespaceException,
			TimeoutException {
		return readNextEvent(timeout);
	}

	public Tree readNextEvent(long timeout) throws TuplespaceException,
			TimeoutException {
		// System.out.println("readNextEvent called");
		InnerNode temp = new InnerNode(queueElemName, id, new LeafNode(Long
				.toString(nextEventIndexToRead)), new LeafNode("?"));
		// System.out.println("now call readIfExists on tuple "+temp);
		InnerNode ret = (InnerNode) space.readIfExists(temp);
		// System.out.println("readIfExists returned with "+ret);
		if (ret == null) {
			// System.out.println("ret not found");
			InnerNode queue = (InnerNode) space.read(queueQueryTuple, timeout);
			if (queue == null)
				throw new TimeoutException("queue tuple read timed out");

			long end = Long.parseLong(((LeafNode) queue.getSubTree(3))
					.getUTF8Data());

			// System.out.println("queue end = "+end);

			while (ret == null && nextEventIndexToRead <= end) {
				nextEventIndexToRead++;
				((LeafNode) temp.getSubTree(2)).setUTF8Data(Long
						.toString(nextEventIndexToRead));
				ret = (InnerNode) space.readIfExists(temp);
			}
		}

		// System.out.println("new nextEventIndexToRead = "+nextEventIndexToRead);

		if (ret != null) {
			nextEventIndexToRead++;
			return ret.getSubTree(3);
		}

		return null;
	}

	/**
	 * adds new element to the end of the queue with specified leaseTime
	 * 
	 * @param tuple
	 *            element to add
	 * @param timeout
	 *            time span to wait trying to take queue tuple
	 * @param leaseTime
	 *            time to live for inserted tuple
	 * @return element index (may be used by derived classes)
	 * @throws TuplespaceException
	 *             error while reading or writing to tuplespace server
	 * @throws TimeoutException
	 *             timed out while trying to take queue tuple
	 */
	/*
	 * @Override public long push(SExpression tuple, long timeout, long
	 * leaseTime) throws TuplespaceException, TimeoutException { Struct queue =
	 * (Struct) space.take(queueQueryTuple, timeout); if (queue == null) throw
	 * new TimeoutException("queue take timed out");
	 * 
	 * long start = Long.parseLong(((Atom)
	 * queue.getSubExpression(3)).getUTF8Data()); long end =
	 * Long.parseLong(((Atom) queue.getSubExpression(3)).getUTF8Data());
	 * 
	 * // set start index to first still existing event Struct temp = new
	 * Struct(queueElemName, id, new Atom(Long .toString(start)), new
	 * Atom("?")); Struct elem = null; do { if (start > end) break;
	 * 
	 * ((Atom) temp.getSubExpression(2)).setUTF8Data(Long.toString(start)); elem
	 * = (Struct) space.readIfExists(temp);
	 * 
	 * if(elem==null) start++; } while (elem == null);
	 * 
	 * 
	 * end+=1; temp = new Struct(queueElemName, id, new Atom(Long
	 * .toString(end)), tuple); space.write(temp, leaseTime);
	 * 
	 * ((Atom) queue.getSubExpression(2)).setUTF8Data(Long.toString(start));
	 * ((Atom) queue.getSubExpression(3)).setUTF8Data(Long.toString(end));
	 * space.write(queue, this.leaseTime);
	 * 
	 * return end; }
	 */
}
